package com.google.code.garbagecan.cassandrastudy.astyanax;

import com.netflix.astyanax.AstyanaxContext;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.NodeDiscoveryType;
import com.netflix.astyanax.connectionpool.OperationResult;
import com.netflix.astyanax.connectionpool.impl.ConnectionPoolConfigurationImpl;
import com.netflix.astyanax.connectionpool.impl.CountingConnectionPoolMonitor;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.model.*;
import com.netflix.astyanax.serializers.StringSerializer;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;

import java.util.Calendar;

public class TestAstyanax {

	private static final String HOST = "perf3";
	private static final int PORT = 9160;
	private static final String CLUSTER_NAME = "benchmark";
	private static final String KEY_SPACE = "demo";
	private static final String TABLE_NAME = "users";

	public static void main(String[] args) throws Exception {
		TestAstyanax ta = new TestAstyanax();
		ta.testInsert();
		ta.testSelect();
		ta.testFind();
	}

	private void testInsert() throws Exception {
		AstyanaxContext<Keyspace> context = getContext();
		context.start();

		Keyspace keyspace = context.getClient();
		MutationBatch m = keyspace.prepareMutationBatch();

		ColumnFamily<String, String> users =
				new ColumnFamily<String, String>(
						TABLE_NAME,
						StringSerializer.get(),
						StringSerializer.get());

		m.withRow(users, "user_" + System.currentTimeMillis())
				.putColumn("firstname", "Fanbin", null)
				.putColumn("lastname", "Kong", null)
				.putColumn("address", "zzzzzzzzzz", null)
				.putColumn("birthday", Calendar.getInstance().getTime(), null);

		OperationResult<Void> result = m.execute();

		context.shutdown();
	}

	private void testSelect() throws Exception {
		AstyanaxContext<Keyspace> context = getContext();
		context.start();

		Keyspace keyspace = context.getClient();
		ColumnFamily<String, String> users =
				new ColumnFamily<String, String>(
						TABLE_NAME,
						StringSerializer.get(),
						StringSerializer.get());

		// Get all users
		Rows<String, String> rows = keyspace.prepareQuery(users).getAllRows().execute().getResult();
		for (Row<String, String> row : rows) {
			System.out.println(row.getKey());
			ColumnList<String> columns = row.getColumns();
			for (Column<String> column : columns) {
				if (column.getName().equals("birthday")) {
					System.out.println(column.getName() + ":" + column.getDateValue());
				} else {
					System.out.println(column.getName() + ":" + column.getStringValue());
				}
			}
		}

		context.shutdown();
	}

	private void testFind() throws Exception {
		AstyanaxContext<Keyspace> context = getContext();
		context.start();

		Keyspace keyspace = context.getClient();
		ColumnFamily<String, String> users =
				new ColumnFamily<String, String>(
						TABLE_NAME,
						StringSerializer.get(),
						StringSerializer.get());

		// Find user
		ColumnList<String> columns = keyspace.prepareQuery(users).getKey("1").execute().getResult();
		for (Column<String> column : columns) {
			if (column.getName().equals("birthday")) {
				System.out.println(column.getName() + ":" + column.getDateValue());
			} else {
				System.out.println(column.getName() + ":" + column.getStringValue());
			}
		}

		context.shutdown();
	}

	private AstyanaxContext getContext() {
		AstyanaxContext<Keyspace> context = new AstyanaxContext.Builder()
				.forCluster(CLUSTER_NAME)
				.forKeyspace(KEY_SPACE)
				.withAstyanaxConfiguration(new AstyanaxConfigurationImpl()
						.setDiscoveryType(NodeDiscoveryType.RING_DESCRIBE)
						.setCqlVersion("3.0.0")
						.setTargetCassandraVersion("1.2")
				)
				.withConnectionPoolConfiguration(new ConnectionPoolConfigurationImpl("MyConnectionPool")
						.setPort(PORT)
						.setMaxConnsPerHost(1)
						.setSeeds(HOST+":"+PORT)
				)
				.withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
				.buildKeyspace(ThriftFamilyFactory.getInstance());

		return context;
	}
}
